package com.alinesno.cloud.base.logger.consumer;

import java.util.Date;
import java.util.Optional;

import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;
import com.alinesno.cloud.base.logger.entity.LogSystemExceptionEntity;
import com.alinesno.cloud.base.logger.service.ILogSystemExceptionService;
import com.alinesno.cloud.common.facade.wrapper.RestWrapper;
import com.alinesno.cloud.monitor.agent.aspect.BaseAspect;

/**
 * 异常消息监听
 * 
 * @author WeiXiaoJin
 * @since 2020年1月25日 下午19:45:08
 */
@Component
class MontorMethodExceptionConsumerListener extends BaseConsumer {

	private static final Logger log = LoggerFactory.getLogger(MontorMethodExceptionConsumerListener.class);
	
	private static final int MAX_MESSAGE_STACK_LENGTH = 2000 ; 
	private static final int MAX_MESSAGE_LENGTH = 512 ; 

	@Autowired
	private ILogSystemExceptionService logSystemExceptionService;

	/**
	 * 服务异常监听
	 * 
	 * @param record
	 * @param data
	 * @param partition
	 * @param topic
	 * @param ts
	 */
	@KafkaListener(topics = { BaseAspect.WATCHER_TOPIC_METHOD_EXCEPTION })
	public void listenExceptionLogger(ConsumerRecord<?, ?> record, 
			@Payload String data,
			@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
			@Header(KafkaHeaders.RECEIVED_TOPIC) String topic, 
			@Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts) {
		
		log.debug(">>>>>>>>>>>>>>>>>>>>>>> 接收异常信息 >>>>>>>>>>>>>>>>>>>>");

		Optional<?> kafkaMessage = Optional.ofNullable(record.value());
		if (kafkaMessage.isPresent()) {

			Object message = kafkaMessage.get();

			if (message != null) {
				JSONObject json = JSONObject.parseObject(message + "");
				
				String methodKey = json.getString("methodKey") ; // 方法唯一值
				LogSystemExceptionEntity e = logSystemExceptionService.findByMethodKey(methodKey) ; 
			
				if(e == null) {
					e = new LogSystemExceptionEntity();
					
					String applicationName = json.getString(APPLICATIONNAME);
					String ip = json.getString(IP);
					String hostname = json.getString(HOSTNAME);

					String className = json.getString("className");
					long createTime = json.getLongValue("createTime");
					String methodName = json.getString("methodName");
					String methodParams = json.getString("methodParams");
					long methodTime = json.getLongValue("methodTime");
					
					e.setApplicationName(applicationName);
					e.setIp(ip);
					e.setHostname(hostname);
					e.setCreateTime(new Date(createTime));
					e.setAddTime(new Date(createTime));
					e.setMethodTime(methodTime);
					e.setMethod(methodName);
					e.setMethodDesc(className);
					e.setParams(methodParams);
					e.setErrorCount(1); 
					e.setMethodKey(methodKey);
				} 
				
				String errorMessage = json.getString("errorMessage");
				String errorStack = json.getString("errorStack");
		
				if(StringUtils.isNoneBlank(errorStack) && errorStack.length()> MAX_MESSAGE_STACK_LENGTH	) {
					errorStack = errorStack.substring(0 , MAX_MESSAGE_STACK_LENGTH) ;
				}
				e.setErrorStack(errorStack);
				
				if(StringUtils.isNoneBlank(errorMessage) && errorMessage.length()> MAX_MESSAGE_LENGTH	) {
					errorMessage = errorMessage.substring(0 , MAX_MESSAGE_LENGTH) ;
				}
				e.setErrorMessage(errorMessage);
				
				// id设置为零，然后保存为新的数据
				e.setId(null); 

				long count = logSystemExceptionService.count(RestWrapper.create().eq("methodKey", methodKey).toSpecification()) ; 
				System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> count = " + count);
				e.setErrorCount((int)count+1);
				
				logSystemExceptionService.save(e);
			}

		}
	}
}
